package com.pengheng.vehicle;

import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.RandomUtil;
import org.apache.commons.lang3.ThreadUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class VehicleParkingAnalysis {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟车辆数据输入流
        DataStream<Tuple3<String, Integer, Long>> vehicleDataStream = env.addSource(new SourceFunction<Tuple3<String, Integer, Long>>() {
            private boolean isRunning = true;

            @Override
            public void run(SourceContext<Tuple3<String, Integer, Long>> ctx) throws Exception {
                //图书的排行榜
//                List<String> plateNoList = Arrays.asList("粤AFQ8900","湘F92888");
                List<String> plateNoList = generateRandomLicensePlates(20);
                while (isRunning) {
                    long timestamp = System.currentTimeMillis() / 1000 * 1000;
                    for (String plateNo : plateNoList) {
                        ctx.collect(new Tuple3<>(plateNo, RandomUtil.randomInt(0, 9) * 10, timestamp));
                    }
                    ThreadUtils.sleep(Duration.ofSeconds(1));
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        // 将数据转换为车辆对象
        DataStream<VehicleData> vehicleObjectStream = vehicleDataStream.map(new MapFunction<Tuple3<String, Integer, Long>, VehicleData>() {
            @Override
            public VehicleData map(Tuple3<String, Integer, Long> tuple3) {
                return new VehicleData(tuple3.f0, tuple3.f1, tuple3.f2);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner((SerializableTimestampAssigner<VehicleData>) (element, recordTimestamp) -> element.getTimestamp()));

        // 按车牌分组
        KeyedStream<VehicleData, String> keyedStream = vehicleObjectStream.keyBy(VehicleData::getLicensePlate);

        // 计算停车次数和停车时长
        DataStream<Tuple4<String, Tuple2<Integer, Long>, Long, Long>> _15sStream = keyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(15)))
                .process(new ParkingAnalyzer());
//                .flatMap(new ParkingAnalyzer());

        // 输出结果
        _15sStream.printToErr("5秒数据统计");
        _15sStream.sinkTo(FileSink.<Tuple4<String, Tuple2<Integer, Long>, Long, Long>>forRowFormat(new Path("f:/tmp"),
                        new SimpleStringEncoder<>())
                .withOutputFileConfig(OutputFileConfig.builder()
                        .withPartPrefix("5秒钟数据")
                        .withPartSuffix(".log")
                        .build())
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd", ZoneId.systemDefault()))
                .withRollingPolicy(DefaultRollingPolicy
                        .builder()
                        .withRolloverInterval(Duration.ofMinutes(1))
                        .withMaxPartSize(new MemorySize(1024 * 1024))
                        .build())
                .build()
        ).setParallelism(1);

        // 计算停车次数和停车时长
        DataStream<Tuple4<String, Tuple2<Integer, Long>, Long, Long>> _1mStream = keyedStream
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .process(new ParkingAnalyzer());
        _1mStream.printToErr("1分钟数据统计");

        _1mStream.sinkTo(FileSink.<Tuple4<String, Tuple2<Integer, Long>, Long, Long>>forRowFormat(new Path("f:/tmp"),
                        new SimpleStringEncoder<>())
                .withOutputFileConfig(OutputFileConfig.builder()
                        .withPartPrefix("1分钟数据")
                        .withPartSuffix(".log")
                        .build())
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd", ZoneId.systemDefault()))
                .withRollingPolicy(DefaultRollingPolicy
                        .builder()
                        .withRolloverInterval(Duration.ofMinutes(1))
                        .withMaxPartSize(new MemorySize(1024))
                        .build())
                .build()
        ).setParallelism(1);
        env.execute();
    }

    static class ParkingAnalyzer extends ProcessWindowFunction<VehicleData, Tuple4<String, Tuple2<Integer, Long>, Long, Long>, String, TimeWindow> {

        private ValueState<Long> lastTimestampState;
        private ValueState<Integer> parkingCountState;
        private ValueState<Long> parkingDurationState;

        /**
         * @param context The context to which the window is being evaluated
         * @throws Exception
         */
        @Override
        public void clear(ProcessWindowFunction<VehicleData, Tuple4<String, Tuple2<Integer, Long>, Long, Long>, String, TimeWindow>.Context context) throws Exception {
            super.clear(context);
            lastTimestampState.clear();
            parkingCountState.clear();
            parkingDurationState.clear();
        }

        @Override
        public void open(Configuration parameters) {


            ValueStateDescriptor<Long> lastTimestampDescriptor = new ValueStateDescriptor<>("lastTimestamp", Long.class);
            lastTimestampState = getRuntimeContext().getState(lastTimestampDescriptor);

            ValueStateDescriptor<Integer> parkingCountDescriptor = new ValueStateDescriptor<>("parkingCount", Integer.class);
            parkingCountState = getRuntimeContext().getState(parkingCountDescriptor);

            ValueStateDescriptor<Long> parkingDurationDescriptor = new ValueStateDescriptor<>("parkingDuration", Long.class);
            parkingDurationState = getRuntimeContext().getState(parkingDurationDescriptor);
        }


        /**
         * @param s
         * @param context
         * @param iterable
         * @param collector
         * @throws Exception
         */
        /**
         * Evaluates the window and outputs none or several elements.
         *
         * @param s        The key for which this window is evaluated.
         * @param context  The context in which the window is being evaluated.
         * @param elements The elements in the window being evaluated.
         * @param out      A collector for emitting elements.
         * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
         */
        @Override
        public void process(String s, ProcessWindowFunction<VehicleData, Tuple4<String, Tuple2<Integer, Long>, Long, Long>, String, TimeWindow>.Context context,
                            Iterable<VehicleData> elements, Collector<Tuple4<String, Tuple2<Integer, Long>, Long, Long>> out) throws Exception {
            //收集到 2s内的数据
            long start = context.window().getStart();
            long end = context.window().getEnd();

//            System.out.println("JSON->" + (end - start) / 1000 + "s内的数据" + JSONUtil.toJsonStr(elements) + "窗口时间为" + start + ":" + end);

            elements.forEach(value -> {
                try {
                    Long lastTimestamp = lastTimestampState.value() == null ? 0L : lastTimestampState.value();
                    Integer parkingCount = parkingCountState.value() == null ? 0 : parkingCountState.value();
                    Long parkingDuration = parkingDurationState.value() == null ? 0L : parkingDurationState.value();
                    //如果当前车速为0，并且上一次的车速不为0，表示产生了停车
                    if (value.getSpeed() == 0) {
                        //如果设置过停车时间这次不记录
                        if (lastTimestampState.value() == null) {
                            lastTimestampState.update(value.getTimestamp());
                        }
                    }
                    //有记录停车时间，并且这次的速度不为0
                    if (lastTimestampState.value() != null && value.getSpeed() != 0) {
                        //停车次数+1
                        parkingCount++;
                        //根据上次的停车时间获取到停车时长
                        parkingDuration += value.getTimestamp() - lastTimestamp;
                        parkingCountState.update(parkingCount);
                        parkingDurationState.update(parkingDuration);
                        //清空停车时间信息
                        lastTimestampState.clear();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            out.collect(Tuple4.of(s, Tuple2.of(ObjUtil.defaultIfNull(parkingCountState.value(), 0), ObjUtil.defaultIfNull(parkingDurationState.value(), 0L)), start, end));
        }
    }

    static class VehicleData {
        private String licensePlate;
        private long speed;
        private long timestamp;

        public VehicleData(String licensePlate, long speed, long timestamp) {
            this.licensePlate = licensePlate;
            this.speed = speed;
            this.timestamp = timestamp;
        }

        public String getLicensePlate() {
            return licensePlate;
        }

        public long getSpeed() {
            return speed;
        }

        public long getTimestamp() {
            return timestamp;
        }
    }

    public static List<String> generateRandomLicensePlates(int count) {
        List<String> licensePlates = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            String province = getRandomProvince();
            String letterPart = generateRandomLetters(1);
            String numberPart = generateRandomNumbers(5);
            licensePlates.add(province + letterPart + numberPart);
        }

        return licensePlates;
    }

    public static String getRandomProvince() {
        String[] provinces = {"粤", "京", "沪", "浙", "苏", "鲁", "闽", "川", "渝", "冀", "晋", "辽", "吉", "黑", "皖", "赣", "豫", "鄂", "湘", "桂", "琼", "贵", "云", "陕", "甘", "青", "宁", "新"};
        Random random = new Random();
        return provinces[random.nextInt(provinces.length)];
    }

    public static String generateRandomLetters(int length) {
        StringBuilder sb = new StringBuilder();
        Random random = new Random();

        for (int i = 0; i < length; i++) {
            char c = (char) (random.nextInt(26) + 'A');
            sb.append(c);
        }

        return sb.toString();
    }

    public static String generateRandomNumbers(int length) {
        StringBuilder sb = new StringBuilder();
        Random random = new Random();

        for (int i = 0; i < length; i++) {
            int num = random.nextInt(10);
            sb.append(num);
        }

        return sb.toString();
    }

}